package com.flitsneak.redis.consumer;

import com.flitsneak.common.entity.Info;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;


/**
 * @author flitsneak
 * @email flitsneak@gmail.com
 * @date 2022/8/10 15:11
 * @description
 */
@Slf4j
@RequiredArgsConstructor
@Component
public class RedisConsumer implements StreamListener<String, ObjectRecord<String, Info>> {

    private static final String STREAM_GROUP = "good-group";

    private static final String STREAM_KEY = "good-stream";

    private static final String CONSUMER_A = "consumer-a";

    private final ReactiveRedisTemplate<String, String> reactiveRedisTemplate;

    @SneakyThrows
    @Override
    public void onMessage(ObjectRecord<String, Info> message) {
        RecordId recordId = message.getId();
        Info info = message.getValue();
        log.info("redis stream 接收到消息：{}", CONSUMER_A + "---->" + "消息id："+recordId+"消息内容："+ info);
        //手动消费确认
        log.info("消息id：{}", recordId);
        reactiveRedisTemplate
                .opsForStream()
                .acknowledge(STREAM_KEY, STREAM_GROUP, recordId)
                .subscribe(c -> {
                            log.info("手动签收消息成功：{}", recordId);
                            reactiveRedisTemplate.opsForStream().delete(STREAM_KEY, recordId).subscribe(d -> log.info("删除消息成功"));
                        }
                );
    }
}
